package com.credithc.signature.core.configure.rabbitmq;

import com.credithc.signature.core.rabbitmq.RabbitMQProcess;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.beans.factory.config.ConfigurableBeanFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Scope;
import org.springframework.retry.policy.SimpleRetryPolicy;
import org.springframework.retry.support.RetryTemplate;

/**
 * 描述：rabbitmq配置<br/>
 * 作者：sunyang
 * 创建：2019/5/27 15:35
 * 版本：v1.0.0<br/>
 */
@Slf4j
@Configuration
@Data
public class CoreRabbitConfig {

    @Value("${spring.rabbitmq.host}")
    private String host;

    @Value("${spring.rabbitmq.port}")
    private int port;

    @Value("${spring.rabbitmq.username}")
    private String username;

    @Value("${spring.rabbitmq.password}")
    private String password;

    @Value("${spring.rabbitmq.virtual-host:/}")
    private String virtualHost;

    @Value("${spring.rabbitmq.core.customConfig.exchangeName:signature-core-exchange}")
    private String coreExchangeName;

    @Value("${spring.rabbitmq.core.customConfig.queueName:signature-core-queue}")
    private String coreQueueName;

    @Value("${spring.rabbitmq.core.customConfig.vipExchangeName:signature-core-exchange-vip}")
    private String coreVipExchangeName;

    @Value("${spring.rabbitmq.core.customConfig.vipQueueName:signature-core-queue-vip-1}")
    private String coreVip1QueueName;

    @Value("${spring.rabbitmq.core.customConfig.vip2QueueName:signature-core-queue-vip-2}")
    private String coreVip2QueueName;

    @Value("${spring.rabbitmq.core.customConfig.vip3QueueName:signature-core-queue-vip-3}")
    private String coreVip3QueueName;

    @Value("${spring.rabbitmq.core.customConfig.vip4QueueName:signature-core-queue-vip-4}")
    private String coreVip4QueueName;

    @Value("${spring.rabbitmq.core.customConfig.vip5QueueName:signature-core-queue-vip-5}")
    private String coreVip5QueueName;

    @Value("${spring.rabbitmq.core.customConfig.maxConcurrentConsumers:3}")
    private int maxConcurrentConsumers;

    @Value("${spring.rabbitmq.cache.connection.size:30}")
    private int connectionCacheSize;

    @Value("${spring.rabbitmq.cache.channel.size:25}")
    private int channelCacheSize;

    @Value("${spring.rabbitmq.callBack.customConfig.exchangeName:signature-callback-exchange}")
    private String callBackExchangeName;

    @Value("${spring.rabbitmq.callBack.customConfig.queueName:signature-callback-queue}")
    private String callBackQueueName;

    @Value("${spring.rabbitmq.callBack.customConfig.vipExchangeName:signature-callback-exchange-vip}")
    private String vipCallBackExchangeName;

    @Value("${spring.rabbitmq.callBack.customConfig.vipQueueName:signature-callback-queue-vip}")
    private String vipCallBackQueueName;

    @Value("${spring.rabbitmq.callBack.customConfig.customConfig:100000}")
    private long sendTimeOut;

    @Value("${spring.rabbitmq.callBack.publisher-confirms:true}")
    private boolean publisherConfirms;

    @Value("${spring.rabbitmq.callBack.publisher-returns:true}")
    private boolean publisherReturns;

    @Value("${spring.rabbitmq.callBack.template.retry.max-attempts:3}")
    private int maxAttempts;

    @Value("${spring.rabbitmq.core.channel.prefetchCount:10}")
    private int prefetchCount;

    public static final String RABBITTEMPLATE_NAME = "rabbitTemplate";

    @Bean
    public ConnectionFactory connectionFactory() {
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory(host,port);
        connectionFactory.setUsername(username);
        connectionFactory.setPassword(password);
        connectionFactory.setVirtualHost(virtualHost);
        connectionFactory.setPublisherConfirms(publisherConfirms);
        connectionFactory.setPublisherReturns(publisherReturns);
        connectionFactory.setCacheMode(CachingConnectionFactory.CacheMode.CONNECTION);
        connectionFactory.setChannelCacheSize(channelCacheSize);
        connectionFactory.setConnectionCacheSize(connectionCacheSize);
        connectionFactory.setConnectionLimit(connectionCacheSize);
        return connectionFactory;
    }

    @Bean
    public RabbitAdmin rabbitAdmin(){
        RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory());
        rabbitAdmin.setAutoStartup(true);
        return rabbitAdmin;
    }

    @Bean
    public Queue coreQueue(RabbitAdmin rabbitAdmin) {
        Queue queue = new Queue(coreQueueName, true, false, false);
        rabbitAdmin.declareQueue(queue);
        return queue;
    }
    @Bean
    public Queue coreVip1Queue(RabbitAdmin rabbitAdmin) {
        Queue queue = new Queue(coreVip1QueueName, true, false, false);
        rabbitAdmin.declareQueue(queue);
        return queue;
    }

    @Bean
    public Queue coreVip2Queue(RabbitAdmin rabbitAdmin) {
        Queue queue = new Queue(coreVip2QueueName, true, false, false);
        rabbitAdmin.declareQueue(queue);
        return queue;
    }

    @Bean
    public Queue coreVip3Queue(RabbitAdmin rabbitAdmin) {
        Queue queue = new Queue(coreVip3QueueName, true, false, false);
        rabbitAdmin.declareQueue(queue);
        return queue;
    }

    @Bean
    public Queue coreVip4Queue(RabbitAdmin rabbitAdmin) {
        Queue queue = new Queue(coreVip4QueueName, true, false, false);
        rabbitAdmin.declareQueue(queue);
        return queue;
    }

    @Bean
    public Queue coreVip5Queue(RabbitAdmin rabbitAdmin) {
        Queue queue = new Queue(coreVip5QueueName, true, false, false);
        rabbitAdmin.declareQueue(queue);
        return queue;
    }

    @Bean
    public DirectExchange coreVipExchange(RabbitAdmin rabbitAdmin) {
        DirectExchange directExchange = new DirectExchange(coreVipExchangeName, true, false);
        rabbitAdmin.declareExchange(directExchange);
        return directExchange;
    }

    @Bean
    public Binding coreVip1Binding(RabbitAdmin rabbitAdmin) {
        Queue queue = coreVip1Queue(rabbitAdmin);
        Binding binding = BindingBuilder.bind(queue).to(coreVipExchange(rabbitAdmin)).with(queue.getName());
        rabbitAdmin.declareBinding(binding);
        return binding;
    }

    @Bean
    public Binding coreVip2Binding(RabbitAdmin rabbitAdmin) {
        Queue queue = coreVip2Queue(rabbitAdmin);
        Binding binding = BindingBuilder.bind(queue).to(coreVipExchange(rabbitAdmin)).with(queue.getName());
        rabbitAdmin.declareBinding(binding);
        return binding;
    }

    @Bean
    public Binding coreVip3Binding(RabbitAdmin rabbitAdmin) {
        Queue queue = coreVip3Queue(rabbitAdmin);
        Binding binding = BindingBuilder.bind(queue).to(coreVipExchange(rabbitAdmin)).with(queue.getName());
        rabbitAdmin.declareBinding(binding);
        return binding;
    }

    @Bean
    public Binding coreVip4Binding(RabbitAdmin rabbitAdmin) {
        Queue queue = coreVip4Queue(rabbitAdmin);
        Binding binding = BindingBuilder.bind(queue).to(coreVipExchange(rabbitAdmin)).with(queue.getName());
        rabbitAdmin.declareBinding(binding);
        return binding;
    }

    @Bean
    public Binding coreVip5Binding(RabbitAdmin rabbitAdmin) {
        Queue queue = coreVip5Queue(rabbitAdmin);
        Binding binding = BindingBuilder.bind(queue).to(coreVipExchange(rabbitAdmin)).with(queue.getName());
        rabbitAdmin.declareBinding(binding);
        return binding;
    }

    @Bean
    public DirectExchange coreExchange(RabbitAdmin rabbitAdmin) {
        DirectExchange directExchange = new DirectExchange(coreExchangeName, true, false);
        rabbitAdmin.declareExchange(directExchange);
        return directExchange;
    }

    @Bean
    public Binding coreBinding(RabbitAdmin rabbitAdmin) {
        Queue queue = coreQueue(rabbitAdmin);
        Binding binding = BindingBuilder.bind(queue).to(coreExchange(rabbitAdmin)).with(queue.getName());
        rabbitAdmin.declareBinding(binding);
        return binding;
    }


    @Bean
    public Queue callBackQueue(RabbitAdmin rabbitAdmin) {
        Queue queue = new Queue(callBackQueueName, true, false, false);
        rabbitAdmin.declareQueue(queue);
        return queue;
    }

    @Bean
    public DirectExchange callBackExchange(RabbitAdmin rabbitAdmin) {
        DirectExchange directExchange = new DirectExchange(callBackExchangeName, true, false);
        rabbitAdmin.declareExchange(directExchange);
        return directExchange;
    }

    @Bean
    public Binding callBinding(RabbitAdmin rabbitAdmin) {
        Queue queue = callBackQueue(rabbitAdmin);
        Binding binding = BindingBuilder.bind(queue).to(callBackExchange(rabbitAdmin)).with(queue.getName());
        rabbitAdmin.declareBinding(binding);
        return binding;
    }

    @Bean
    public Queue vipCallBackQueue(RabbitAdmin rabbitAdmin) {
        Queue queue = new Queue(vipCallBackQueueName, true, false, false);
        rabbitAdmin.declareQueue(queue);
        return queue;
    }

    @Bean
    public DirectExchange vipCallBackExchange(RabbitAdmin rabbitAdmin) {
        DirectExchange directExchange = new DirectExchange(vipCallBackExchangeName, true, false);
        rabbitAdmin.declareExchange(directExchange);
        return directExchange;
    }

    @Bean
    public Binding vipCallBinding(RabbitAdmin rabbitAdmin) {
        Queue queue = vipCallBackQueue(rabbitAdmin);
        Binding binding = BindingBuilder.bind(queue).to(vipCallBackExchange(rabbitAdmin)).with(queue.getName());
        rabbitAdmin.declareBinding(binding);
        return binding;
    }


    //-----------------------------------------延迟队列相关----------------------------------------------
    /**
     * 发送到该队列的message会在一段时间后过期进入到core_process_queue
     * 每个message可以控制自己的失效时间
     */

    private String CORE_DELAY_QUEUE_NAME = "core_delay_queue";

    private String CORE_DELAY_EXCHANGE_NAME = "core_delay_exchange";

    private String CORE_PROCESS_QUEUE_NAME = "core_process_queue";

    private String CORE_PROCESS_EXCHANGE_NAME = "core_process_exchange";

    private String CORE_ROUTING_KEY = "coreRoutingKey";


    /**
     * 延迟队列 exchange
     *
     * @return
     */
    @Bean
    public DirectExchange coreDelayExchange() {
        return new DirectExchange(CORE_DELAY_EXCHANGE_NAME);
    }

    /**
     * 正常队列 exchange
     *
     * @return
     */
    @Bean
    public DirectExchange coreProcessExchange() {
        return new DirectExchange(CORE_PROCESS_EXCHANGE_NAME);
    }

    /**
     * 存放延迟消息的队列 最后将会转发给exchange(实际消费队列对应的)
     *
     * @return
     */
    @Bean
    Queue coreDelayQueueForMsg() {
        return QueueBuilder.durable(CORE_DELAY_QUEUE_NAME)
                // DLX，dead letter发送到的exchange
                .withArgument("x-dead-letter-exchange", CORE_PROCESS_EXCHANGE_NAME)
                // dead letter携带的routing key
                .withArgument("x-dead-letter-routing-key", CORE_ROUTING_KEY)
                .build();
    }

    /**
     * 正常消费队列
     *
     * @return
     */
    @Bean
    public Queue coreProcessQueue() {
        return QueueBuilder.durable(CORE_PROCESS_QUEUE_NAME).build();
    }

    /**
     * 将延迟队列与exchange绑定，即到达指定时间之后需要转交给queue消费
     *
     * @return
     */
    @Bean
    Binding coreDelayBinding() {
        return BindingBuilder.bind(coreDelayQueueForMsg()).to(coreDelayExchange()).with(CORE_ROUTING_KEY);
    }

    /**
     * 绑定正常消费队列和exchange
     *
     * @return
     */
    @Bean
    Binding coreQueueBinding() {
        return BindingBuilder.bind(coreProcessQueue()).to(coreProcessExchange()).with(CORE_ROUTING_KEY);
    }

    //---------------------------------------------------------------------------------------------------------


    @Bean
    public MessageConverter jsonMessageConverter() {
        return new Jackson2JsonMessageConverter();
    }

    @Bean
    public SimpleMessageListenerContainer messageContainer(RabbitMQProcess rabbitMQProcess) {
        return getSimpleMessageListenerContainer(rabbitMQProcess,coreQueue(rabbitAdmin()));
    }

    @Bean
    public SimpleMessageListenerContainer messageVip1Container(RabbitMQProcess rabbitMQProcess) {
        return getSimpleMessageListenerContainer(rabbitMQProcess,coreVip1Queue(rabbitAdmin()));
    }

    @Bean
    public SimpleMessageListenerContainer messageVip2Container(RabbitMQProcess rabbitMQProcess) {
        return getSimpleMessageListenerContainer(rabbitMQProcess,coreVip2Queue(rabbitAdmin()));
    }

    @Bean
    public SimpleMessageListenerContainer messageVip3Container(RabbitMQProcess rabbitMQProcess) {
        return getSimpleMessageListenerContainer(rabbitMQProcess,coreVip3Queue(rabbitAdmin()));
    }

    @Bean
    public SimpleMessageListenerContainer messageVip4Container(RabbitMQProcess rabbitMQProcess) {
        return getSimpleMessageListenerContainer(rabbitMQProcess,coreVip4Queue(rabbitAdmin()));
    }

    @Bean
    public SimpleMessageListenerContainer messageVip5Container(RabbitMQProcess rabbitMQProcess) {
        return getSimpleMessageListenerContainer(rabbitMQProcess,coreVip5Queue(rabbitAdmin()));
    }

    private SimpleMessageListenerContainer getSimpleMessageListenerContainer(ChannelAwareMessageListener rabbitMQProcess,Queue queue) {
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory());
        //container.setQueues(coreQueue(rabbitAdmin()));
        container.setQueueNames(queue.getName());
        // 需要将channel暴露给listener才能手动确认，AcknowledgeMode.MANUAL时必须为ture
        container.setExposeListenerChannel(true);
        container.setMaxConcurrentConsumers(maxConcurrentConsumers);
        container.setConcurrentConsumers(maxConcurrentConsumers);
        //设置确认模式手工确认
        container.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        container.setMessageListener(rabbitMQProcess);
        container.setPrefetchCount(prefetchCount);
        // 设为true，扔回queue头部，设为false，丢弃
        container.setDefaultRequeueRejected(true);
        container.setMessageConverter(jsonMessageConverter());
        return container;
    }


    /**
     * 通讯问题导致发送失败的重发模板
     * @return
     */
    @Bean
    public RetryTemplate callBackRetryTemplate() {
        RetryTemplate retryTemplate = new RetryTemplate();
        retryTemplate.setRetryPolicy(new SimpleRetryPolicy(maxAttempts));
        return retryTemplate;
    }

    @Bean(name = RABBITTEMPLATE_NAME)
    @Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
    public RabbitTemplate rabbitTemplate() {
        RabbitTemplate template = new RabbitTemplate(connectionFactory());
        template.setMandatory(true);
        template.setMessageConverter(jsonMessageConverter());
        template.setRetryTemplate(callBackRetryTemplate());
        return template;
    }

}
